package com.atguigu1.core.operator

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 *
 * @description: 相同的key对value进行聚合操作
 * @time: 2021-03-12 11:45
 * @author: baojinlong
 **/
object Spark20GroupByAndGroupByKey {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("rdd")
    // 设置rdd分区数字
    val sparkContext = new SparkContext(conf)
    // 排序后分区数量不变 hash range分区 python分区是私有的
    val rddValue: RDD[(String, Int)] = sparkContext.makeRDD(Seq(("a", 1), ("b", 1), ("a", 2), ("b", 1), ("d", 1), ("a", 3)), 2)

    /**
     * groupByKey:将数据中的数据相同的key的数据分在一个组中,形成一个对偶元组:
     * 元组中的第一个元素就是key
     * 元组中第二个元素就是相同key的value的集合
     */
    // (b,CompactBuffer(1, 1))
    val groupByKeyResult: RDD[(String, Iterable[Int])] = rddValue.groupByKey()
    groupByKeyResult.collect.foreach(println)
    println("****************")
    // (b,CompactBuffer((b,1), (b,1)))
    val groupByResult: RDD[(String, Iterable[(String, Int)])] = rddValue.groupBy(_._1)
    groupByResult.collect.foreach(println)

    /**
     * reduceByKey和groupByKey区别:
     * 1.groupByKey会导致数据打乱重组,存在shuffle操作,spark中shuffle结果必须进行落盘性能比较低下
     * 2.reduceByKey是对相同的key的value两两进行聚合操作.所以在同一个分区内就可以对相同的key进行预聚合combine,可以有效减少shuffle落盘数据量
     * 3.reduceByKey要求分区内和分区间的计算规则是相同的.如果不相同则需要使用aggregateByKey
     */
    sparkContext.stop()
  }
}
